package com.etlmaster.executor.scheduler;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.etlmaster.executor.agent.Agents;
import com.etlmaster.executor.agent.DefaultAgent;
import com.etlmaster.executor.scheduler.utils.SubmitResult;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.LogWriter;


/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class ScheduleServer {

	public volatile static Map<String, Agents> agents = new HashMap<String, Agents>();

	private static List<Map<String, String>> eventMaps = new ArrayList<Map<String, String>>();
	private static Lock lock = new ReentrantLock();// 锁对象
	private static Lock eventLock = new ReentrantLock();// 锁对象
	private static Condition agentNotEmpty = lock.newCondition();

	private static Condition notEmpty = eventLock.newCondition();

	private ScheduleServer() {

	}

	public static SubmitResult submitTask(Proc proc) {
		SubmitResult result = null;
		lock.lock();
		if (agents.size() > 0) {
			int condition = runConditionCheck(proc);
			if (condition == ScheduleStatus.conditionMeet) {

				Agents agent = getAgentForTask(proc);
				if (agent == null) {
					LogWriter.addLog("WARN", "未找到满足条件agent！");
				}else
					result = agent.run(proc);
			} else {
				result = new SubmitResult();
				result.setExecuteResult(ScheduleStatus.conditionNotMeet);
				result.setReturnMsg(ScheduleStatus.statusMap.get(condition));
				LogWriter.addLog("WARN", "{},{},不满足运行条件", proc.getProc_name(),
						ScheduleStatus.statusMap.get(condition));
			}

		} else {
			LogWriter.addLog("WARN", "当前集群未注册任何agent，添加默认agent");

			addDefaultAgent();
			result = submitTask(proc);
		}
		lock.unlock();
		return result;
	}

	private static Agents getAgentForTask(Proc proc) {
		if (proc.getAgentId() != null && proc.getAgentId().trim().length() > 0
				&& !proc.getAgentId().equalsIgnoreCase( "null")) {
			if (agents.get(proc.getAgentId()) == null) {
				LogWriter.addLog("ERROR","No such Agent "
						+ proc.getAgentId() + "!");
				return null;
			} else {
				Agents agent = agents.get(proc.getAgentId());
				if (agent.getMaxThreadNum() - agent.getRunningProcs().size() > 0) {
					return agent;
				} else
					return null;
			}
		} else {

			return getLowestAgent();
		}
	}

	private static Agents getLowestAgent() {
		int lowestBurden = 0;// 最大空余线程数
		String lowestAgentId = "";
		Iterator it = agents.entrySet().iterator();

		while (it.hasNext()) {
			Entry entry = (Entry) it.next();
			Agents agent = (Agents) entry.getValue();
			if (agent.getMaxThreadNum() - agent.getRunningProcs().size() > lowestBurden) {
				lowestBurden = agent.getMaxThreadNum()
						- agent.getRunningProcs().size();
				lowestAgentId = (String) entry.getKey();
			}
		}
		LogWriter.addLog("INFO", "当前负荷最小agent 为{}，剩余任务数为{}", lowestAgentId,
				lowestBurden + "");
		return agents.get(lowestAgentId);
	}

	private static int runConditionCheck(Proc proc) {
		Agents assignedagent = agents.get(proc.getAgentId());
		if (assignedagent == null) {
			Iterator it = agents.entrySet().iterator();
			while (it.hasNext()) {
				Entry entry = (Entry) it.next();
				Agents agent = (Agents) entry.getValue();
				int hasProc = agent.hasProc(proc);
				if (hasProc > ScheduleStatus.sameNameProc) {
					return ScheduleStatus.alreadyRunning;

				} else if (!proc.isConcurrencyRun()
						&& hasProc > ScheduleStatus.noProc) {
					return ScheduleStatus.preCycleRunning;
				}
			}
		} else {
			int hasProc = assignedagent.hasProc(proc);
			if (hasProc > ScheduleStatus.sameNameProc) {
				return ScheduleStatus.alreadyRunning;

			} else if (!proc.isConcurrencyRun()
					&& hasProc > ScheduleStatus.noProc) {
				return ScheduleStatus.preCycleRunning;
			}
		}
		return ScheduleStatus.conditionMeet;
	}

	public static List<Map<String, String>> getEventMaps() {
		return eventMaps;
	}

	public static void setEventMaps(List<Map<String, String>> eventMaps) {
		ScheduleServer.eventMaps = eventMaps;
	}

	public static Lock getLock() {
		return lock;
	}

	public static void setLock(Lock lock) {
		ScheduleServer.lock = lock;
	}

	public static Condition getNotEmpty() {
		return notEmpty;
	}

	public static Lock getEventLock() {
		return eventLock;
	}

	public static void setEventLock(Lock eventLock) {
		ScheduleServer.eventLock = eventLock;
	}

	public static void addDefaultAgent() {
		Agents agent = new DefaultAgent("default", "default",
				DataCollectionUtils.getIntProperties("agentMaxThread"));
		ScheduleServer.agents.put("default", agent);
	}

	public static Condition getAgentNotEmpty() {
		return agentNotEmpty;
	}

	public static void setAgentNotEmpty(Condition agentNotEmpty) {
		ScheduleServer.agentNotEmpty = agentNotEmpty;
	}

}
